package org.apache.flink.connector.scylladb.table;

import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connector.scylladb.config.ScyllaDBConfigOptions;
import org.apache.flink.connector.scylladb.config.ScyllaDBConnectorConstants;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.types.DataType;

import java.util.List;
import java.util.Set;

import java.util.HashSet;
import java.util.Set;

import static org.apache.flink.table.factories.FactoryUtil.createTableFactoryHelper;

/**
 * ScyllaDB 动态表目标工厂，用于创建 ScyllaDB 表目标
 */
public class ScyllaDBDynamicTableSinkFactory implements DynamicTableSinkFactory {

    @Override
    public String factoryIdentifier() {
        return ScyllaDBConnectorConstants.IDENTIFIER;
    }

    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        final Set<ConfigOption<?>> options = new HashSet<>();
        options.add(ScyllaDBConfigOptions.URL);
        options.add(ScyllaDBConfigOptions.KEY_SPACE);
        options.add(ScyllaDBConfigOptions.TABLE_NAME);
        return options;
    }

    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        final Set<ConfigOption<?>> options = new HashSet<>();
        options.add(ScyllaDBConfigOptions.USERNAME);
        options.add(ScyllaDBConfigOptions.PASSWORD);
        options.add(ScyllaDBConfigOptions.SINK_PARALLELISM);
        options.add(ScyllaDBConfigOptions.SINK_BUFFER_FLUSH_MAX_ROWS);
        options.add(ScyllaDBConfigOptions.SINK_BUFFER_FLUSH_INTERVAL);
        options.add(ScyllaDBConfigOptions.SINK_MAX_RETRIES);
        options.add(ScyllaDBConfigOptions.SINK_APPEND_SET_FIELDS);
        options.add(ScyllaDBConfigOptions.CONNECTION_POOL_LOCAL_SIZE);
        options.add(ScyllaDBConfigOptions.CONNECTION_POOL_REMOTE_SIZE);
        options.add(ScyllaDBConfigOptions.CONNECTION_TIMEOUT);
        options.add(ScyllaDBConfigOptions.REQUEST_TIMEOUT);
        options.add(ScyllaDBConfigOptions.CONSISTENCY_LEVEL);
        options.add(ScyllaDBConfigOptions.SSL_ENABLED);
        options.add(ScyllaDBConfigOptions.SSL_KEYSTORE_PATH);
        options.add(ScyllaDBConfigOptions.SSL_KEYSTORE_PASSWORD);
        options.add(ScyllaDBConfigOptions.SSL_TRUSTSTORE_PATH);
        options.add(ScyllaDBConfigOptions.SSL_TRUSTSTORE_PASSWORD);
        return options;
    }

    @Override
    public DynamicTableSink createDynamicTableSink(Context context) {
        // 使用 FactoryUtil 帮助器验证选项并获取配置
        final FactoryUtil.TableFactoryHelper helper = createTableFactoryHelper(this, context);
        
        // 验证所有选项
        helper.validate();
        
        // 获取配置
        final ReadableConfig config = helper.getOptions();
        
        // 获取物理数据类型
        final DataType physicalDataType = context.getCatalogTable().getResolvedSchema().toPhysicalRowDataType();
        
        // 解析 Set 追加字段配置
        String appendSetFieldsStr = config.get(ScyllaDBConfigOptions.SINK_APPEND_SET_FIELDS);
        Set<String> appendSetFields = new java.util.HashSet<>();
        if (appendSetFieldsStr != null && !appendSetFieldsStr.trim().isEmpty()) {
            String[] fields = appendSetFieldsStr.split(",");
            for (String field : fields) {
                appendSetFields.add(field.trim());
            }
        }
        
        // 提取主键信息
        List<String> primaryKeys = new java.util.ArrayList<>();
        if (context.getCatalogTable().getResolvedSchema().getPrimaryKey().isPresent()) {
            primaryKeys = context.getCatalogTable().getResolvedSchema().getPrimaryKey().get().getColumns();
        }
        
        // 创建并返回 ScyllaDB 表目标
        return new ScyllaDBDynamicTableSink(
                config.get(ScyllaDBConfigOptions.URL),
                config.get(ScyllaDBConfigOptions.KEY_SPACE),
                config.get(ScyllaDBConfigOptions.TABLE_NAME),
                config.get(ScyllaDBConfigOptions.USERNAME),
                config.get(ScyllaDBConfigOptions.PASSWORD),
                config.get(ScyllaDBConfigOptions.SINK_PARALLELISM),
                config.get(ScyllaDBConfigOptions.SINK_BUFFER_FLUSH_MAX_ROWS),
                config.get(ScyllaDBConfigOptions.SINK_BUFFER_FLUSH_INTERVAL),
                config.get(ScyllaDBConfigOptions.SINK_MAX_RETRIES),
                config.get(ScyllaDBConfigOptions.CONNECTION_POOL_LOCAL_SIZE),
                config.get(ScyllaDBConfigOptions.CONNECTION_POOL_REMOTE_SIZE),
                config.get(ScyllaDBConfigOptions.CONNECTION_TIMEOUT),
                config.get(ScyllaDBConfigOptions.REQUEST_TIMEOUT),
                config.get(ScyllaDBConfigOptions.CONSISTENCY_LEVEL),
                config.get(ScyllaDBConfigOptions.SSL_ENABLED),
                config.getOptional(ScyllaDBConfigOptions.SSL_KEYSTORE_PATH).orElse(null),
                config.getOptional(ScyllaDBConfigOptions.SSL_KEYSTORE_PASSWORD).orElse(null),
                config.getOptional(ScyllaDBConfigOptions.SSL_TRUSTSTORE_PATH).orElse(null),
                config.getOptional(ScyllaDBConfigOptions.SSL_TRUSTSTORE_PASSWORD).orElse(null),
                physicalDataType,
                appendSetFields,
                primaryKeys
        );
    }
} 